package com.gzhryc.socket.netty;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import org.apache.commons.lang3.StringUtils;

import com.gzhryc.socket.SocketCallback;
import com.gzhryc.socket.SocketConfig;
import com.gzhryc.socket.SocketMessage;
import com.gzhryc.socket.SocketNode;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.bytes.ByteArrayDecoder;
import io.netty.handler.codec.bytes.ByteArrayEncoder;

public class NettyThreadPoolTCPClient extends SocketNode {

    private ThreadPoolExecutor executor;
    private BlockingQueue<Runnable> blockQueue;
    private MainThread mainThread;

    SocketCallback mCallback;

    private Channel mChannel;

    public NettyThreadPoolTCPClient(SocketConfig config, SocketCallback callback) {
        this.config = config;
        this.mCallback = callback;

        this.blockQueue = new LinkedBlockingQueue<Runnable>();
        this.executor = new ThreadPoolExecutor(config.getCorePoolSize(), config.getMaximumPoolSize()
                , config.getKeepAliveTime(), TimeUnit.MILLISECONDS, this.blockQueue);
    }

    @Override
    public void start() {
        final NettyThreadPoolTCPClient self = this;
        //设置一个多线程循环器
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        //启动附注类
        Bootstrap bootstrap = new Bootstrap();
        bootstrap.group(workerGroup);
        //指定所使用的NIO传输channel
        bootstrap.channel(NioSocketChannel.class);
        //指定客户端初始化处理
        bootstrap.handler(new ChannelInitializer<SocketChannel>() {
            @Override
            protected void initChannel(SocketChannel socketChannel) throws Exception {
                //获取管道
                ChannelPipeline pipeline = socketChannel.pipeline();
                //字符串解码器
                pipeline.addLast(new ByteArrayDecoder());
                //字符串编码器
                pipeline.addLast(new ByteArrayEncoder());
                //处理类
                pipeline.addLast(new ByteNettyClientChannelHandler(self, mCallback));
            }
        });
        try {
            //连接服务
            mChannel = bootstrap.connect(config.getIp(), config.getPort()).sync().channel();
            mChannel.closeFuture().sync();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            workerGroup.shutdownGracefully();
        }
    }

    @Override
    public void startService() {
        if (mainThread == null || !mainThread.isAlive()) {
            mainThread = new MainThread(this);
            mainThread.setDaemon(true);        //守护线程
            mainThread.start();
        }
    }

    public void runTask(Runnable runnable) {
        this.executor.submit(runnable);
    }

    @Override
    public boolean writeAndFlush(SocketMessage message) {
        if (mChannel != null && mChannel.isWritable()) {
            mChannel.writeAndFlush(message.getBytes());
            return true;
        }
        return false;
    }

    @Override
    public void close() {
        if (mChannel != null) {
            try {
                mChannel.close().sync();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    public static class MainThread extends Thread {

        final NettyThreadPoolTCPClient mClient;

        public MainThread(NettyThreadPoolTCPClient client) {
            this.mClient = client;
        }

        public void run() {
            mClient.start();
        }
    }

    @Override
    public boolean isConnected() {
        if (this.mChannel != null) {
            return mChannel.isActive();
        }
        return false;
    }

    @Override
    public String getId() {
        if (this.mChannel != null) {
            if (StringUtils.isNotBlank(config.getClientId())) {
                return config.getClientId();
            } else {
                return mChannel.id().asLongText();
            }
        }
        return null;
    }

    @Override
    public void reconnect() {
        if (mainThread != null) {
            mainThread.interrupt();
        }
        this.blockQueue.clear();  //清空任务队列
        if (mainThread != null) {
            //重新启动
            mainThread = new NettyThreadPoolTCPClient.MainThread(this);
            mainThread.setDaemon(true); //守护线程
            mainThread.start();
        } else {
            start();
        }

    }
}
